<?
define('ROOT', $_SERVER['DOCUMENT_ROOT']);
require_once(ROOT.'/public/global.php');
require_once(ROOT.'/lang/'.$LN.'.php');
?>
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
<? require_once(ROOT.'/public/ref.php'); ?>
<title>mac-message - 进阶</title>
<meta name="description" content="mac-message activemq jms broker 消息中间件 消息中心" />
<meta name="keywords" content="mac-message activemq jms broker 消息中间件 消息中心 消息监控" />
<script type="text/javascript">
$(function(){
	function adjust(){
		var de = document.documentElement
			, ch = de.clientHeight, cw = de.clientWidth;
		$('#body').width(cw - 16);
	}
	$(window).resize(adjust);
	adjust();
});
</script>
</head>
<body>
<? require_once(ROOT.'/public/head.php'); ?>
<div id="body">
	<div class="dir">
<? require_once('./dir.php'); ?>
	</div>
	<div class="main">
		<div id="nav">
			<div id="nav_path">
				<a href="<?=$LP?>">Home</a>
				&gt;
				<a href="<?=$LP?>mac/message">mac-message</a>
				&gt;
				进阶
			</div>
			<div id="nav_back">
				<a href="javascript:history.back()"><?=$Msg['back']?></a>
				|
				<a href="<?=$LP?>"><?=$Msg['up']?></a> &nbsp;&nbsp;
			</div>
			<div class="clear"></div>
		</div>
<h1>mac-message - 进阶</h1>
<h2><a name="converter"></a>自定义消息转换器</h2>
&nbsp;&nbsp;框架中预定义了两种消息转换器。JmsMsgConverter和AmqMsgConverter，可自由添加和实现。<br/>
&nbsp;&nbsp;JmsMsgConverter 消息转换器用于实现 javax.jms.Message 与 com.boarsoft.message.bean.Message 之间的互转。<br/>
&nbsp;&nbsp;ActiveMQ长连接客户端会使用JmsMsgConverter实现两种消息的转换。 
<xmp>
	public class DemoJmsMsgConverter implements JmsMsgConverter {
	
		@Override
		public Message convert(javax.jms.Message am) throws Exception {
			String o = ((TextMessage) am).getText();
			Message m = JsonUtil.parseObject(o, Message.class);
			return m;
		}
	
		@Override
		public javax.jms.Message convert(Session session, Message m) throws Exception {
			// return session.createTextMessage((String) m.getContent());
			return session.createTextMessage(m.toJson());
		}
	}
</xmp>
&nbsp;&nbsp;AmqMsgConverter 消息转换器用于实现 org.apache.activemq.command.Message 与 com.boarsoft.message.bean.Message 之间的互转。<br/>
&nbsp;&nbsp;消息中间件的ActiveMQ插件会使用AmqMsgConverter实现两种消息的转换。 
<xmp>
	public class AmqJsonMsgConverter implements AmqMsgConverter {
		private static final Logger log = LoggerFactory.getLogger(AmqJsonMsgConverter.class);
	
		@Override
		public com.boarsoft.message.bean.Message convert(Message am) throws Exception {
			ActiveMQTextMessage tm = (ActiveMQTextMessage) am;
			String s = tm.getText();
			log.debug("Received activemq text message: {}", s);
			return JsonUtil.parseObject(s, com.boarsoft.message.bean.Message.class);
		}
	
		@Override
		public Message convert(com.boarsoft.message.bean.Message m) throws Exception {
			return null;
		}
	}
</xmp>
<h2><a name="interceptor"></a>ActiveMQ消息拦截与转换</h2>
&nbsp;&nbsp;拦截ActiveMQ消息并转换为内部的消息处理方式，这样就可以实现ActiveMQ与基于RPC的生产者和消费者的互通。<br/>
&nbsp;&nbsp;中间件的 activemq.xml<br/>
<xmp>
	<broker xmlns="http://activemq.apache.org/schema/core" useJmx="false">
		<transportConnectors>
			<transportConnector uri="amqp://0.0.0.0:5672" />
			<transportConnector uri="tcp://0.0.0.0:61616" />
		</transportConnectors>
		<plugins>
			<discardingDLQBrokerPlugin dropAll="true"
				dropTemporaryTopics="true" dropTemporaryQueues="true" />

			<!-- ActiveMQ消息拦截器 -->
			<bean xmlns="http://www.springframework.org/schema/beans" class="com.boarsoft.message.amq.plugin.InterceptorPlugin">
				<property name="localMsgSvc1" ref="localMsgSvc1" />
				<!-- 消息转换器列表，用于将ActiveMQ消息转换为mac-message中定义的消息对象 -->
				<property name="converterMap">
					<map>
						<!-- Key为ActiveMQ中的队列名 -->
						<entry key="queue1">
							<bean class="com.boarsoft.boar.message.demo.AmqJsonMsgConverter" />
						</entry>
					</map>
				</property>
			</bean>
		</plugins>
	</broker>
</xmp>
<h2><a name="ack"></a>收件人消息签收</h2>
&nbsp;&nbsp;消息中间件通过向发件人发送签收回执消息来实现消息的签收。按照常理，一般是由消息发送者负责接收消息签收的回执，因此编写一个发件人来发送消息，并接收回执。
<xmp>
	@Component("queue1Producer1")
	public class Queue1Producer1 implements MessageService, MessageSender {
		private static final Logger log = LoggerFactory.getLogger(Queue1Producer1.class);
	
		@Override
		public void put(Message m) throws Exception {
			log.info("Received acknowledge: {}", JsonUtil.from(m));
		}
	
		@Override
		public void send(Message m) throws Exception {
			// 设置from作为签收回执消息的消费者
			m.setFrom("queue1Producer1");
			Messenger.send(m);
		}
	}
</xmp>
&nbsp;&nbsp;让生产者暴露这个服务，provide.xml
<xmp>
	<service ref="queue1Producer1" group="message" name="queue1Producer1"
		interface="com.boarsoft.message.core.MessageService"
		version="1.0.0" timeout="12000">
	</service>
</xmp>
&nbsp;&nbsp;让中间件引用这个服务，consume.xml
<xmp>
	<reference id="queue1Producer1" group="message" name="queue1Producer1"
		interface="com.boarsoft.message.core.MessageService" version="1.0.0"
		timeout="12000">
	</reference>
</xmp>
&nbsp;&nbsp;发送消息时先设置消息的Acknowledge属性再调它的send方法即可（注意此方法有设置from属性）。
<xmp>
	final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:conf/context.xml");
	Message m = new Message();
	m.setTarget("queue1");
	m.setContent("Hello world!");
	m.setId(RandomUtil.genUUID());
	m.addRecipient("queue1Consumer1");
	// 填写用于接收签收回执的消息队列
	m.setAcknowledge("ack1");
	Queue1Producer1 qp1 = (Queue1Producer1) ctx.getBean("queue1Producer1");
	qp1.send(m);
</xmp>
<h2><a name="finder"></a>过滤与分拣插件</h2>
&nbsp;&nbsp;用于动态决定每条消息的收件人。实现如下接口即可：
<xmp>
	/**
	 * 返回收件人列表，其中key为收件人ID，value为是否必达
	 * 
	 * @param m
	 * @return
	 */
	public interface RecipientFinder {
		Map<String, Boolean> find(Message m);
	}
</xmp>
	</div>
</div>
<? require_once(ROOT.'/public/foot.html'); ?>
</body>
</html>
<?
$fp = ROOT.$LP.'/mac/message/';
output(basename(__FILE__), $fp.'advance.html');
?>